package sparkSql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.junit.{After, Before, Test}

class Transform {
	val conf: SparkConf = new SparkConf().setAppName("sparkSql").setMaster("local[3]")

	var outpath: String = "out"

	val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

	import util.MyPredef._

	@Before
	def init() {
		outpath.delete()
	}

	@After
	def after() {
		spark.stop()
	}

	/**
	 * 1、隐士转换 import spark.
	 */
	@Test
	def RDDToDf(): Unit = {
		val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1, "zhangsan", 20), (2, "lisi", 30), (3, "wangwu", 40)))

		import spark.implicits._

		val df: DataFrame = rdd.toDF("id", "name", "age")
		df.show()

		val person: RDD[Person] = rdd.map { case (id, name, age) => Person(id, name, age) }
		person.toDF().show()

	}

	/**
	 * 1、隐士转换 import spark.
	 */
	@Test
	def RDDToDS(): Unit = {
		val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1, "zhangsan", 20), (2, "lisi", 30), (3, "wangwu", 40)))

		import spark.implicits._

		val person: RDD[Person]  = rdd.map{
			case (id, name, age) => Person(id, name, age)
		}

		val ds: Dataset[Person] = person.toDS()

		ds.show()


	}

	/**
	 * 1、
	 */
	@Test
	def DFAToDS(): Unit = {
		val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1, "zhangsan", 20), (2, "lisi", 30), (3, "wangwu", 40)))

		import spark.implicits._

		val df: DataFrame = rdd.toDF("id", "name", "age")
		val ds: Dataset[Person] = df.as[Person]
		ds.show()


		ds.toDF().show()
		val rdd1: RDD[Person] = ds.rdd
		val rdd2: RDD[Row] = ds.toDF().rdd

		rdd2.foreach{
			case row => println(row.getInt(0), row.getString(1), row.getInt(2))
		}

	}

	/**
	 * 1、
	 */
	@Test
	def DSToDF(): Unit = {
		val df: DataFrame = spark.read.json("in/user.json")
		import spark.implicits._
		val ds: Dataset[Person2] = df.as[Person2]

		ds.toDF().show()

		val rdd1: RDD[Row] = ds.toDF().rdd
		rdd1.foreach(println)

		val rdd2: RDD[Person2] = ds.rdd
		rdd2.foreach(println)
	}


}

case class Person(id: Int, name: String, age: Int)

case class Person2(name: String, age: Long)